package com.doit.day01;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class ProducerDemo {
    public static void main(String[] args) throws InterruptedException {


        Properties props = new Properties();

        //这三个参数是必须配置的
        props.setProperty("bootstrap.servers","linux01:9092,linux02:9092,linux03:9092");
        //静态常量
        props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"linux01:9092,linux02:9092,linux03:9092");

        props.setProperty("key.serialzer",StringSerializer.class.getName());
        props.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
        props.setProperty("value.serializer",StringSerializer.class.getName());
        //选配的
        //分区器
        props.setProperty("partitioner.class",MyPartitioner.class.getName());
        //kafka生产者应答机制
        props.setProperty("acks","-1");
        //生产者的对象
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        //对消息进行封装

        for (int i = 4; i < 5; i++) {
            ProducerRecord<String, String> record = new ProducerRecord<>("ideatest", "smessage from java"+i);
            //拿着对象去send消息
            producer.send(record);

            producer.flush();
        }



        //关闭资源
        producer.close();

    }
}





